package io.kiki.sba.registry.server.metadata.slot.util.builder;


import io.kiki.sba.registry.common.model.slot.Slot;
import io.kiki.sba.registry.exception.SofaRegistryRuntimeException;
import io.kiki.sba.registry.util.DatumVersionUtil;
import lombok.Getter;
import org.apache.commons.lang.StringUtils;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;


public class SlotBuilder implements Builder<Slot> {

    @Getter
    private final int slotId;
    private final Set<String> followerDataStoreNodeIdSet = new HashSet<>();
    private final int followerNums;
    @Getter
    private long leaderEpoch;
    @Getter
    private String leaderDataStoreNodeId;

    public SlotBuilder(int slotId, int followerNums) {
        this.slotId = slotId;
        this.followerNums = followerNums;
    }

    public SlotBuilder(int slotId, String leaderDataStoreNodeId, long leaderEpoch, int followerNums) {
        this.slotId = slotId;
        this.leaderDataStoreNodeId = leaderDataStoreNodeId;
        this.leaderEpoch = leaderEpoch;
        this.followerNums = followerNums;
    }

    public boolean addFollowerDataStoreNodeId(Set<String> followerDataStoreNodeIdSet) {
        for (String followerDataStoreNodeId : followerDataStoreNodeIdSet) {
            if (!addFollowerDataStoreNodeId(followerDataStoreNodeId)) {
                return false;
            }
        }
        return true;
    }

    public boolean addFollowerDataStoreNodeId(String followerDataStoreNodeId) {
        if (StringUtils.isBlank(followerDataStoreNodeId)) {
            throw new IllegalArgumentException("add empty follower");
        }
        if (followerDataStoreNodeIdSet.contains(followerDataStoreNodeId)) {
            return true;
        }
        if (followerDataStoreNodeIdSet.size() >= followerNums) {
            return false;
        }
        followerDataStoreNodeIdSet.add(followerDataStoreNodeId);
        return true;
    }


    public boolean removeFollowerDataStoreNodeId(String followerDataStoreNodeId) {
        return followerDataStoreNodeIdSet.remove(followerDataStoreNodeId);
    }

    public int getFollowerDataStoreNodeIdSetSize() {
        return followerDataStoreNodeIdSet.size();
    }

    public boolean containsFollowerDataStoreNodeId(String followerDataStoreNodeId) {
        return followerDataStoreNodeIdSet.contains(followerDataStoreNodeId);
    }

    private boolean isReady() {
        return leaderDataStoreNodeId != null;
    }

    public SlotBuilder setLeaderDataStoreNodeId(String leaderDataStoreNodeId) {
        if (!StringUtils.equals(this.leaderDataStoreNodeId, leaderDataStoreNodeId)) {
            this.leaderDataStoreNodeId = leaderDataStoreNodeId;
            leaderEpoch = DatumVersionUtil.nextId();
        }
        return this;
    }

    public Set<String> getFollowerDataStoreNodeIdSet() {
        return Collections.unmodifiableSet(new HashSet<>(followerDataStoreNodeIdSet));
    }

    @Override
    public Slot build() {
        if (!isReady()) {
            throw new SofaRegistryRuntimeException("slot builder is not ready for build: leader[" + leaderDataStoreNodeId + "], followers[" + StringUtils.join(followerDataStoreNodeIdSet, ",") + "]");
        }
        if (leaderEpoch <= 0) {
            this.leaderEpoch = DatumVersionUtil.nextId();
        }
        return new Slot(slotId, leaderDataStoreNodeId, leaderEpoch, followerDataStoreNodeIdSet);
    }


}
